#![allow(unused_imports)]

use reqwest ;
use serde_json ;
use ring::hmac ;
// use hex ;
use base64 ;
use uuid::Uuid ;

use chrono::prelude::*;
use std::time::{SystemTime, UNIX_EPOCH} ;
use std::sync::{Arc, Mutex};

use std::net::SocketAddr;
use std::collections::{HashSet,HashMap} ;

use urlencoding ;

use rdkafka::client::ClientContext;
use rdkafka::consumer::{CommitMode, Consumer, ConsumerContext, Rebalance};
use rdkafka::error::KafkaResult;
use rdkafka::topic_partition_list::TopicPartitionList;
use rdkafka::consumer::stream_consumer::StreamConsumer;

use rdkafka::message::{Message} ;
use rdkafka::config::ClientConfig;
use rdkafka::producer::{FutureProducer, FutureRecord};

use schema_registry_converter::async_impl::schema_registry::SrSettings;
use schema_registry_converter::async_impl::avro::AvroDecoder ;

use futures::future::{ready};
use futures::future ;
use futures::stream::{StreamExt};

use bigdecimal::BigDecimal;
use num_traits::{FromPrimitive, Num, One, Signed, ToPrimitive, Zero};
use std::str::FromStr;

use dotenv::dotenv;
use std::env::var ;

use cdrs::authenticators::NoneAuthenticator;
use cdrs::cluster::session::{new, Session};
use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder};
use cdrs::load_balancing::RoundRobin;
use cdrs::query::*;
use cdrs::query_values;
use cdrs::types::tuple::Tuple;
use cdrs::types::{AsRust, ByName, IntoRustByName};

use avro_rs::types::Value as AvroValue ;

use std::net::Ipv4Addr ;

use utils::* ;

mod utils {
    use super::* ;

    pub fn now_ts() -> u128 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards").as_millis()
    }
    pub fn now_iso_ts_s() -> String {
        let utc: DateTime<Utc> = Utc::now();
        utc.format("%Y-%m-%dT%H:%M:%SZ").to_string()
    }
    pub fn generate_once() -> String {
        let my_uuid_s = Uuid::new_v4().to_string();
        let now_ts_s = now_ts().to_string() ;
        format!("{}{}", my_uuid_s, now_ts_s)
    }

    pub fn hmac_sha_sign(secret: &str, msg: &str) -> String {
        let key = hmac::Key::new(hmac::HMAC_SHA1_FOR_LEGACY_USE_ONLY, secret.as_bytes()) ;
        let tag = hmac::sign(&key, msg.as_bytes()) ;
        base64::encode(tag.as_ref())
    }
    pub fn resolve_socket_addr(hosts: &str) -> Option<SocketAddr> {
        let ips = get_if_addrs::get_if_addrs().unwrap().iter()
            .map(|interface| interface.ip()).collect::<HashSet<_>>();

        println!("resolve_socket_addr hosts is  : {}", hosts);
        for ip in ips.iter() {
            println!("resolve_socket_addr get_if_addrs ip: {}", ip);
        }

        hosts.split(",")
            .map(|x| x.replace("http://", ""))
            .filter_map(|x| x.parse::<SocketAddr>().ok())
            .filter(|x| ips.contains(&x.ip()))
            .next()
    }
}

// https://github.com/aliyun/aliyun-openapi-java-sdk/blob/master/aliyun-java-sdk-core/src/main/java/com/aliyuncs/endpoint/LocationServiceEndpointResolver.java
// https://github.com/aliyun/aliyun-openapi-java-sdk/blob/master/aliyun-java-sdk-core/src/main/java/com/aliyuncs/auth/HmacSHA1Signer.java
// https://github.com/aliyun/aliyun-openapi-java-sdk/blob/master/aliyun-java-sdk-geoip/src/main/java/com/aliyuncs/geoip/model/v20200101/DescribeIpv4LocationRequest.java
async fn location_api(access_key_id: &str, access_secret: &str, region_id: &str, service_code: &str) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
    let aliyun_url = "https://location-readonly.aliyuncs.com" ;
    let location_version = "2015-06-12" ;
    let signature_version = "1.0" ;
    let location_params = [("AccessKeyId", access_key_id.into()),
                           ("Action", "DescribeEndpoints".into()),
                           ("Format", "JSON".into()),
                           ("Id", region_id.into()),
                           ("RegionId", region_id.into()),
                           ("ServiceCode", service_code.into()),
                           ("SignatureMethod", "HMAC-SHA1".into()),
                           ("SignatureNonce", generate_once()),
                           ("SignatureVersion", signature_version.into()),
                           ("Timestamp", urlencoding::encode(&now_iso_ts_s())),
                           ("Type", "openAPI".into()),
                           ("Version", location_version.into())
    ] ;

    let location_sign_body = format!("GET&%2F&{}", 
                            urlencoding::encode(&location_params.iter().map(|(k, v)| format!("{}={}", k, v))
                                                .collect::<Vec<_>>().join("&"))
    ) ;
    let location_sign_val = urlencoding::encode(&hmac_sha_sign(&format!("{}&", access_secret), &location_sign_body)) ;

    let url_frags_sign = location_params.iter()
        .chain([("Signature",  location_sign_val)].iter())
        .map(|(k, v)| format!("{}={}", k, v))
        .collect::<Vec<_>>().join("&") ;

    Ok(reqwest::get(&format!("{}?{}", aliyun_url, &url_frags_sign))
        .await?
        .json()
        .await?)
}

async fn geoip_api(access_key_id: &str, access_secret: &str, region_id: &str, geoip_host: &str, ip: &str) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
    let geoip_version = "2020-01-01" ;
    let signature_version = "1.0" ;
    let geoip_params = [("AccessKeyId", access_key_id.to_string()),
                        ("Action", "DescribeIpv4Location".into()),
                        ("Format", "JSON".into()),
                        ("Ip", ip.into()),
                        ("RegionId", region_id.into()),
                        ("ServiceCode", "geoip".into()),
                        ("SignatureMethod", "HMAC-SHA1".into()),
                        ("SignatureNonce", generate_once()),
                        ("SignatureVersion", signature_version.into()),
                        ("Timestamp", urlencoding::encode(&now_iso_ts_s())),
                        ("Version", geoip_version.into()),
    ] ;

    let geoip_sign_body = format!("POST&%2F&{}", 
                            urlencoding::encode(&geoip_params.iter().map(|(k, v)| format!("{}={}", k, v))
                                                .collect::<Vec<_>>().join("&"))
    ) ;

    let geoip_sign_val = urlencoding::encode(&hmac_sha_sign(&format!("{}&", access_secret), &geoip_sign_body)) ;
    let geoip_frags_sign = geoip_params.iter()
        .chain([("Signature",  geoip_sign_val)].iter())
        .map(|(k, v)| format!("{}={}", k, v))
        .collect::<Vec<_>>().join("&") ;

    Ok(reqwest::Client::new()
        .post(&format!("http://{}?{}", geoip_host, &geoip_frags_sign))
        .send()
        .await?
        .json()
        .await?)

}

mod kafka {
    use super::* ;

    pub struct CustomContext;
    impl ClientContext for CustomContext {}
    impl ConsumerContext for CustomContext {
        fn pre_rebalance(&self, _rebalance: &Rebalance) {
        }

        fn post_rebalance(&self, _rebalance: &Rebalance) {
        }

        fn commit_callback(&self, _result: KafkaResult<()>, _offsets: &TopicPartitionList) {
        }
    }

    pub fn mk_sr(hosts: &str) -> SrSettings {
        let mut sr_hosts = hosts.split(",") ;
        let first_sr_host = sr_hosts.next().expect("at least one ACCESS_KAFKA_SR_HOSTS needed!") ;
        let mut sr_settings_builder = SrSettings::new_builder(first_sr_host.into()) ;
        sr_hosts.for_each(|x| {sr_settings_builder.add_url(x.into());}) ;
        sr_settings_builder.build().expect("kafka-sr build error!")
    }
}

mod avro {
    use super::* ;

    pub fn as_map(v: &AvroValue) -> HashMap<String, AvroValue> {
        match v {
            AvroValue::Record(items) => items.to_owned().into_iter().collect(),
            _ => HashMap::new(),
        }
    }

    pub fn as_int(v: &AvroValue) -> Option<i32> {
        match *v {
            AvroValue::Int(i) => Some(i),
            _ => None,
        }
    }
    pub fn as_string(v: &AvroValue) -> Option<String> {
        match v {
            AvroValue::String(s) => Some(s.to_string()),
            _ => None,
        }
    }

}


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {

    dotenv::from_filename("meta-lib-static/env/.env").ok();

    let env_hosts = var("EXCHANGE_GATE_ALIYUN_GEOIP_HOSTS").unwrap_or("127.0.0.1:0".into()) ;

    /**
    if resolve_socket_addr(&env_hosts).is_none() {
        println!("EXCHANGE_GATE_ALIYUN_GEOIP_HOSTS not match! skip") ;
        return Ok(()) ;
    }
    */

    let env_kafkas = var("EXCHANGE_GATE_ALIYUN_GEOIP_IN_KAFKAS").unwrap_or("127.0.0.1:9092".into()) ;
    let env_kafka_prefix = var("EXCHANGE_GATE_ALIYUN_GEOIP_IN_KAFKA_PREFIX").unwrap_or("data_pipeline_dev".into()) ;
    let env_kafka_srs = var("EXCHANGE_GATE_ALIYUN_GEOIP_IN_KAFKA_SRS").unwrap_or("127.0.0.1:8081".into()) ;
    let access_key_id = var("EXCHANGE_GATE_ALIYUN_GEOIP_IN_ALIYUN_API_ID").expect("NO EXCHANGE_GATE_ALIYUN_GEOIP_IN_ALIYUN_API_ID FOUND!") ;
    let access_secret = var("EXCHANGE_GATE_ALIYUN_GEOIP_IN_ALIYUN_API_SECRET").expect("NO EXCHANGE_GATE_ALIYUN_GEOIP_IN_ALIYUN_API_SECRET FOUND!") ;

    let cassandras = var("EXCHANGE_GATE_ALIYUN_GEOIP_OUT_CASSANDRAS").unwrap_or("127.0.0.1:9042".to_string()) ;
    let cassandra_prefix = var("EXCHANGE_GATE_ALIYUN_GEOIP_OUT_CASSANDRA_PREFIX").unwrap_or("data_pipeline_dev".to_string()) ;

    let region_id = "cn-hangzhou".to_string() ;
    let location_reply = location_api(&access_key_id, &access_secret, &region_id, "geoip").await? ;
    let geoip_host = location_reply["Endpoints"]["Endpoint"][0]["Endpoint"].as_str().unwrap() ;

    let decoder = AvroDecoder::new(kafka::mk_sr(&env_kafka_srs));
    let decoder_arc = &Arc::new(Mutex::new(decoder));
    let topic = format!("{}_ip_dt_dim", env_kafka_prefix) ;
    let consumer: StreamConsumer<kafka::CustomContext> = ClientConfig::new()
        .set("group.id", &format!("{}:exchange-aliyun-geoip", topic))
        .set("bootstrap.servers", &env_kafkas)
        .set("auto.offset.reset", "smallest")
        .set("enable.auto.commit", "true")
        .create_with_context(kafka::CustomContext)
        .expect("Consumer creation failed");
    consumer.subscribe(&[topic.as_str()][..]).expect("Can't subscribe to specified topics") ;

    println!("cassandras: {}", cassandras) ;
    let nodes = cassandras.split(",").map(|host| NodeTcpConfigBuilder::new(host, NoneAuthenticator {}).build()).collect::<Vec<_>>() ;
    let cluster_config = ClusterTcpConfig(nodes);
    let cassandra_sess = &new(&cluster_config, RoundRobin::new()).expect("session should be created") ;

    let access_key_id_ref = &access_key_id ;
    let access_secret_ref = &access_secret ;
    let region_id_ref = &region_id ;
    let cassandra_prefix_ref = &cassandra_prefix ;
    println!("topic: {} -> group: {}", topic, format!("{}:exchange-aliyun-geoip", topic)) ;

    consumer.start()
        .filter_map( |x| future::ready( x.ok() ) ) 
        .filter_map( |x| async move {
            let decode_one = decoder_arc.clone() ;
            let mut decode_mut = decode_one.lock().unwrap() ;
            decode_mut.decode(x.payload()).await.ok() 
        }) 
        .filter_map( |x| future::ready( {
            let row = avro::as_map(&x.value) ;
            let ip_if = row.get("ip").map_or_else(|| None, |v| avro::as_string(v)) ;
            let dt_if = row.get("dt").map_or_else(|| None, |v| avro::as_int(v)) ;
            let tm_if = row.get("tm").map_or_else(|| None, |v| avro::as_int(v)) ;

            match (ip_if, dt_if, tm_if) {
                (Some(ip), Some(dt), Some(tm)) => Some((ip, dt, tm)),
                _ => None
            }
        }))
        .for_each (|(ip, dt, _tm)| async move {
            if ip == "0.0.0.0" { return () } ;
            let ip_addr : Ipv4Addr = ip.parse().unwrap() ;
            if ip_addr.is_private() { return ()  } ;
            
            let ip_val = i32::from_be_bytes(ip_addr.octets()) ;
            
            let row = &cassandra_sess.query_with_values(
                format!("select geo_id from {}.ip_dt_dim where ip_id = ? and dt = ?", cassandra_prefix_ref), 
                query_values!(ip_val, dt)
            ).unwrap().get_body().unwrap().into_rows().unwrap()[0] ;
            let geo_id: Option<Tuple> = row.get_by_name("geo_id").expect("geo_id") ;
            if geo_id.is_some() { return () } ;
            let geoip_reply = geoip_api(access_key_id_ref, access_secret_ref, region_id_ref, geoip_host, &ip).await.unwrap() ;

            let latitude_decimal_some = geoip_reply["Latitude"].as_str() ;
            if latitude_decimal_some.is_none() {
                println!("error response: {:?}", geoip_reply) ;
                return () 
            }
            let latitude_decimal = &BigDecimal::from_str(geoip_reply["Latitude"].as_str().unwrap()).unwrap() * BigDecimal::from(1_000_000) ;
            let longitude_decimal = &BigDecimal::from_str(geoip_reply["Longitude"].as_str().unwrap()).unwrap() * BigDecimal::from(1_000_000);
            let latitude_i32 = latitude_decimal.to_i32().unwrap() ;
            let longitude_i32 = longitude_decimal.to_i32().unwrap() ;
            
            let isp = geoip_reply["Isp"].as_str().unwrap() ;

            let cassandra_update = cassandra_sess.query_with_values(
                format!("UPDATE {}.ip_dt_dim SET geo_id = (?, ?), isp = ? WHERE ip_id = ? AND dt = ?", cassandra_prefix_ref),
                query_values!(latitude_i32, longitude_i32, isp, ip_val, dt)
            ) ;
            if cassandra_update.is_err() { 
                println!("result: {:?}", cassandra_update) ;
                return () ;
            } ;
            let cassandra_insert = cassandra_sess.query_with_values(
                format!(
                    "INSERT INTO {}.geo_dim(geo_id, country_code, country_en, country, province_en, province, city_en, city, county)
                     VALUES((?, ?), ?, ?, ?, ?, ?, ?, ?, ?)",
                    cassandra_prefix_ref),
                query_values!(
                    latitude_i32, longitude_i32, 
                    geoip_reply["CountryCode"].as_str().unwrap(), geoip_reply["CountryEn"].as_str().unwrap(), geoip_reply["Country"].as_str().unwrap(),
                    geoip_reply["ProvinceEn"].as_str().unwrap(), geoip_reply["Province"].as_str().unwrap(),
                    geoip_reply["CityEn"].as_str().unwrap(), geoip_reply["City"].as_str().unwrap(), geoip_reply["County"].as_str().unwrap()
                )
            ) ;
            if cassandra_insert.is_err() {
                println!("result: {:?}", cassandra_update) ;
                return () ;
            }
        }).await ;

    Ok(())
}
